package com.xxl.job.admin.core.thread;

import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
import com.xxl.job.admin.core.trigger.TriggerTypeEnum;
import com.xxl.job.admin.core.trigger.XxlJobTrigger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * job trigger thread pool helper
 * 任务触发线程池管理助手
 * <p>
 * 职责：
 * 1. 维护快速/慢速双线程池，根据任务执行情况动态分配线程池
 * 2. 监控任务执行超时情况并自动降级处理
 * 3. 提供任务触发入口及生命周期管理
 *
 * @author xuxueli 2018-07-03 21:08:07
 */
public class JobTriggerPoolHelper {
    private static Logger logger = LoggerFactory.getLogger(JobTriggerPoolHelper.class);


    // ---------------------- trigger pool ----------------------

    // fast/slow thread pool
    /**
     * 快速任务线程池配置：
     * - 核心线程数：10
     * - 最大线程数：从配置获取
     * - 队列容量：2000
     * 用于处理常规快速任务
     */
    private ThreadPoolExecutor fastTriggerPool = null;
    /**
     * 慢速任务线程池配置：
     * - 核心线程数：10
     * - 最大线程数：从配置获取
     * - 队列容量：5000
     * 用于处理降级的慢速任务（1分钟内超时超过10次的任务）
     */
    private ThreadPoolExecutor slowTriggerPool = null;

    /**
     * 启动方法，初始化快速和慢速触发池
     * 该方法用于准备线程池以执行不同的任务根据任务的特性（快速或慢速）进行区分
     */
    public void start() {
        // 初始化快速触发池
        fastTriggerPool = new ThreadPoolExecutor(
                10,
                XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax(),
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(2000),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        // 为快速触发池中的线程命名
                        return new Thread(r, "xxl-job, admin JobTriggerPoolHelper-fastTriggerPool-" + r.hashCode());
                    }
                },
                new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        // 当快速触发池拒绝执行任务时记录错误日志
                        logger.error(">>>>>>>>>>> xxl-job, admin JobTriggerPoolHelper-fastTriggerPool execute too fast, Runnable=" + r.toString());
                    }
                });

        // 初始化慢速触发池
        slowTriggerPool = new ThreadPoolExecutor(
                10,
                XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax(),
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(5000),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        // 为慢速触发池中的线程命名
                        return new Thread(r, "xxl-job, admin JobTriggerPoolHelper-slowTriggerPool-" + r.hashCode());
                    }
                },
                new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        // 当慢速触发池拒绝执行任务时记录错误日志
                        logger.error(">>>>>>>>>>> xxl-job, admin JobTriggerPoolHelper-slowTriggerPool execute too fast, Runnable=" + r.toString());
                    }
                });
    }


    /**
     * 停止线程池服务
     * <p>
     * 执行操作：
     * 1. 立即关闭快速任务线程池
     * 2. 立即关闭慢速任务线程池
     * 3. 记录关闭成功日志
     */
    public void stop() {
        //triggerPool.shutdown();
        fastTriggerPool.shutdownNow();
        slowTriggerPool.shutdownNow();
        logger.info(">>>>>>>>> xxl-job trigger thread pool shutdown success.");
    }


    // job timeout count
    /**
     * 时间片标记（分钟级）
     * 用于每分钟重置超时计数器
     */
    private volatile long minTim = System.currentTimeMillis() / 60000;     // ms > min

    /**
     * 任务超时计数器映射表
     * Key: 任务ID
     * Value: 在最近1分钟内的超时次数
     */
    private volatile ConcurrentMap<Integer, AtomicInteger> jobTimeoutCountMap = new ConcurrentHashMap<>();


    /**
     * 添加触发任务到线程池
     *
     * @param jobId                 任务唯一标识
     * @param triggerType           触发类型枚举（MANUAL/CRON等）
     * @param failRetryCount        失败重试策略：
     *                              >=0 使用当前值作为重试次数
     *                              <0  使用任务配置的默认重试次数
     * @param executorShardingParam 分片执行参数（用于分布式分片场景）
     * @param executorParam         执行参数（覆盖任务默认参数）
     * @param addressList           目标执行器地址列表（多个地址用逗号分隔）
     */
    public void addTrigger(final int jobId,
                           final TriggerTypeEnum triggerType,
                           final int failRetryCount,
                           final String executorShardingParam,
                           final String executorParam,
                           final String addressList) {

        // choose thread pool
        // 根据超时次数动态选择线程池
        ThreadPoolExecutor triggerPool_ = fastTriggerPool;
        AtomicInteger jobTimeoutCount = jobTimeoutCountMap.get(jobId);
        // 当任务1分钟内超时超过10次时切换慢速线程池
        if (jobTimeoutCount != null && jobTimeoutCount.get() > 10) {      // job-timeout 10 times in 1 min
            triggerPool_ = slowTriggerPool;
        }

        // 提交异步触发任务到线程池
        triggerPool_.execute(new Runnable() {
            @Override
            public void run() {

                long start = System.currentTimeMillis();

                try {
                    // 执行核心触发逻辑 do trigger
                    XxlJobTrigger.trigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam, addressList);
                } catch (Throwable e) {
                    logger.error(e.getMessage(), e);
                } finally {
                    // 每分钟重置超时计数器
                    long minTim_now = System.currentTimeMillis() / 60000;
                    if (minTim != minTim_now) {
                        minTim = minTim_now;
                        jobTimeoutCountMap.clear();
                    }

                    // incr timeout-count-map
                    // 记录超时任务计数（耗时超过500ms视为超时）
                    long cost = System.currentTimeMillis() - start;
                    if (cost > 500) {       // ob-timeout threshold 500ms
                        AtomicInteger timeoutCount = jobTimeoutCountMap.putIfAbsent(jobId, new AtomicInteger(1));
                        if (timeoutCount != null) {
                            timeoutCount.incrementAndGet();
                        }
                    }

                }

            }

            @Override
            public String toString() {
                return "Job Runnable, jobId:" + jobId; // 线程池任务标识
            }
        });
    }


    // ---------------------- helper ----------------------
    /**
     * 单例助手实例
     */
    private static JobTriggerPoolHelper helper = new JobTriggerPoolHelper();

    /**
     * 启动线程池服务（静态入口）
     */
    public static void toStart() {
        helper.start();
    }

    /**
     * 停止线程池服务（静态入口）
     */
    public static void toStop() {
        helper.stop();
    }

    /**
     * 触发任务执行（静态入口）
     *
     * @param jobId                 任务唯一标识
     * @param triggerType           触发类型枚举
     * @param failRetryCount        失败重试策略
     *                              >=0: use this param
     *                              <0: use param from job info config
     * @param executorShardingParam 分片参数
     * @param executorParam         null: use job param
     *                              not null: cover job param
     */
    public static void trigger(int jobId, TriggerTypeEnum triggerType, int failRetryCount, String executorShardingParam, String executorParam, String addressList) {
        helper.addTrigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam, addressList);
    }

}
